package com.cloud.easy.im.handler.qos;

import com.cloud.easy.im.argeement.Agreement;
import com.cloud.easy.im.utils.BasicThreadFactory;
import com.cloud.easy.im.utils.SystemClock;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Map;
import java.util.concurrent.*;

/**
 * C2S(对于服务端来说是接受场景)模式中QoS数据包质量包证机制之发送队列保证实现类.
 * <p>
 * 主流的移动端IM都有这个机制，具体流转如下：
 * 假设有A,B两个客户端发送数据并开启了Qos机制，使用了4个包：
 * A->S S->B B->S S->A（准确地说是两个包的两次转发，一个是IM消息包一个是IM应答包
 * <p>
 * note：
 * 由于是接受场景，服务端只需要保存协议体中指纹{@link {@link Agreement#getFp()}和时间戳，确保消息最终到达即可。
 * <p>
 * 本类和{@link QoS4SendDaemonRoot}是QoS机制的核心，目的是加强保证TCP协议在应用层的可靠性和送达率。
 * 当前QoS机制支持全部的C2C、C2S、S2C共3种消息交互场景下的消息送达质量保证：
 * <p>
 * 1)Client to Server(C2S)：即由某客户端主动发起，消息最终接收者是服务端，此模式下：重发由C保证、ACK应答由S发回；
 * 2)Server to Client(S2C)：即由服务端主动发起，消息最终接收者是某客户端，此模式下：重发由S保证、ACK应答由C发回；
 * 3)Client to Client(C2C)：即由客户端主动发起，消息最终接收者是另一客户端。此模式对于QoS机制来说，相当于C2S+S2C两程路径。
 *
 * @version 0.1.0
 * @create 2020年10月10日2:57 下午
 */
public class QoS4RecivedDaemonRoot {

    private static final Logger logger = LoggerFactory.getLogger(QoS4RecivedDaemonRoot.class);
    /**
     * 5分钟：qos队列检查时间
     */
    private int CHECH_INTERVAL = 5 * 60;
    /**
     * 10分钟：qos消息最长保留时长
     */
    private int MESSAGES_VALID_TIME = 10 * 60;
    private final ConcurrentMap<String, Long> recievedMessages = new ConcurrentHashMap<String, Long>();
    /**
     * 本定时任务作用在于定时清理已经超时还没有收到qos的数据
     **/
    private final ScheduledExecutorService reciveDaemon = new ScheduledThreadPoolExecutor(1,
            new BasicThreadFactory.Builder().namingPattern("qos-recive-daemon-pool-%d").daemon(true).build());

    private boolean excuting = false;
    /**
     * TAG for log
     */
    private String TAG = null;

    protected QoS4RecivedDaemonRoot(int chechInterval, int messagesValidTime, String tag) {
        if (chechInterval > 0) {
            this.CHECH_INTERVAL = chechInterval;
        }
        if (messagesValidTime > 0) {
            this.MESSAGES_VALID_TIME = messagesValidTime;
        }
        this.TAG = tag;
    }

    private void doTaskOnece() {
        if (!excuting) {
            excuting = true;
            logger.debug("【Qos4ReciveDaemonRoot" + this.TAG + "】【QoS接收方】++++++++++ START 暂存处理线程正在运行中，当前长度" + recievedMessages.size() + ".");

            for (Map.Entry<String, Long> entry : recievedMessages.entrySet()) {
                String key = entry.getKey();
                long value = entry.getValue();
                long delta = SystemClock.getInstance().getNow() - value;
                if (delta >= MESSAGES_VALID_TIME) {
                    logger.debug("【Qos4ReciveDaemonRoot" + this.TAG + "】【QoS接收方】指纹为" + key + "的包已生存" + delta
                            + "ms(最大允许" + MESSAGES_VALID_TIME + "ms), 马上将之删除.");
                    recievedMessages.remove(key);
                }
            }
        }
        logger.debug("【Qos4ReciveDaemonRoot" + this.TAG + "】【QoS接收方】++++++++++ END 暂存处理线程正在运行中，当前长度" + recievedMessages.size() + ".");
        excuting = false;
    }

    public synchronized void startup() {
        stop();
        if (size() > 0) {
            for (String key : recievedMessages.keySet()) {
                putImpl(key);
            }
        }
        reciveDaemon.scheduleAtFixedRate(this::doTaskOnece, CHECH_INTERVAL, CHECH_INTERVAL, TimeUnit.SECONDS);
    }

    public synchronized void stop() {
        if (reciveDaemon.isShutdown()) {
            reciveDaemon.shutdown();
        }
    }

    /**
     * 添加ack队列
     * <p/>
     * 会自动自动判断是否重复
     *
     * @param agreement
     */
    public boolean addRecieved(Agreement agreement) {
        if (agreement != null && agreement.isQoS()) {
            return addRecieved(agreement.getFp());
        }
        return false;
    }

    public boolean addRecieved(String fingerPrintOfProtocal) {
        if (fingerPrintOfProtocal == null) {
            logger.debug("【Qos4ReciveDaemonRoot" + this.TAG + "】无效的指纹特征码 fingerPrintOfProtocal==null!");
            return false;
        }

        if (hasRecieved(fingerPrintOfProtocal)) {
            logger.debug("【Qos4ReciveDaemonRoot" + this.TAG + "】【QoS接收方】指纹为" + fingerPrintOfProtocal
                    + "的消息已经存在于接收列表中，该消息重复了（原理可能是对方因未收到应答包而错误重传导致），更新收到时间戳哦.");
        }

        return putImpl(fingerPrintOfProtocal);
    }

    private boolean putImpl(String fingerPrintOfProtocal) {
        if (fingerPrintOfProtocal != null) {
            recievedMessages.put(fingerPrintOfProtocal, SystemClock.getInstance().getNow());
            return true;
        }
        return false;
    }

    public boolean hasRecieved(String fingerPrintOfProtocal) {
        return recievedMessages.containsKey(fingerPrintOfProtocal);
    }

    public int size() {
        return recievedMessages.size();
    }
}
